package atguigu.com.edu.app.dwd.log;

import atguigu.com.edu.util.DateFormatUtil;
import atguigu.com.edu.util.MyKafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;


import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import org.apache.commons.lang3.StringUtils;

public class DwdTrafficUniqueVisitorDetail {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

       // env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

/*        env.enableCheckpointing(60000L, CheckpointingMode.EXACTLY_ONCE);

        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(6000L);

        env.getCheckpointConfig().
                enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));

        env.setStateBackend(new HashMapStateBackend());

        env.getCheckpointConfig().setCheckpointStorage("hdfs//hadoop202:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        String topic = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_uv_group";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);

        DataStreamSource<String> stringDataStreamSource = env.addSource(kafkaConsumer);

        SingleOutputStreamOperator<JSONObject> jsonObj = stringDataStreamSource.map(JSON::parseObject);

        KeyedStream<JSONObject, String> jsonKey = jsonObj.keyBy(r -> r.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<JSONObject> filterDs = jsonKey.filter(new RichFilterFunction<JSONObject>() {

            private ValueState<String> last;

            @Override
            public void open(Configuration parameters) throws Exception {

                ValueStateDescriptor<String> lastV = new ValueStateDescriptor<>("last", String.class);

                lastV.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());

                last = getRuntimeContext().getState(lastV);


            }

            @Override
            public boolean filter(JSONObject in) throws Exception {

                String lastPageId = null;
                try {
                    lastPageId = in.getJSONObject("page").getString("last_page_id");
                } catch (Exception e) {

                } finally {
                }

                if (StringUtils.isNotEmpty(lastPageId)){
                    return false;

                }
                String value = last.value();
                String curVisitDate = DateFormatUtil.toDate(in.getLong("ts"));
                if (StringUtils.isEmpty(value) || !value.equals(curVisitDate)){
                    last.update(curVisitDate);
                    return true;
                }
return false;
            }
        });
        filterDs.print(">>>>");

        env.execute();



    }

}
